package com.sdk.es.http.kafka;

import cn.hutool.json.JSONUtil;
import com.sdk.es.http.conts.Conts;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * @author limaosheng
 * @create 2023-08-16 16:20
 * @desc
 */
@Slf4j
@Component
public class EsProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendSave(String index,Object object){
        String topic = Conts.TOPIC_PREFIX + index + Conts.TOPIC_SAVE_SUFFIX;
        this.send(topic,object);
    }

    public void sendUpdate(String index,Object object){
        String topic = Conts.TOPIC_PREFIX + index + Conts.TOPIC_UPDATE_SUFFIX;
        this.send(topic,object);
    }

    public void sendDelete(String index,Object object){
        String topic = Conts.TOPIC_PREFIX + index + Conts.TOPIC_DELETE_SUFFIX;
        this.send(topic,object);
    }

    public void send(String topic,Object object){
        String message = JSONUtil.toJsonStr(object);
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        future.whenComplete((result, exception) -> {
            if (exception == null) {
                log.info("KafkaProducer.send:{} 发送消息成功:{}", topic, message);
            } else {
                log.error("KafkaProducer.send:{} 发送消息失败:{}", topic, exception.getMessage());
            }
        });
    }
}
